package com.fwmagic.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;

public class MyCoprocessor implements RegionObserver,RegionCoprocessor {

    Logger logger = LoggerFactory.getLogger(MyCoprocessor.class);

    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }

    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {
        logger.error("==========>enter Coprocessor");

        NavigableMap<byte[], List<Cell>> fm = put.getFamilyCellMap();
        Map.Entry<byte[], List<Cell>> e = fm.firstEntry();

        List<Cell> cells = e.getValue();
        Cell cell = cells.get(0);
        byte[] uid = CellUtil.cloneRow(cell);
        byte[] s = CellUtil.cloneQualifier(cell);

        logger.error("======>uid:"+uid);
        logger.error("==========>s:"+s);

        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "master:2181,slave1:2181,slave2:2181");
        Connection connection = ConnectionFactory.createConnection(config);

        Table table = connection.getTable(TableName.valueOf("t_fans"));

        Put put1 = new Put(s);
        put1.addColumn("f".getBytes(), uid, uid);
        table.put(put1);
        logger.error("==========>put is over");

        table.close();
        connection.close();
    }
}
